package com.mango.ch05;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class RelativeFrequencyMapper extends Mapper<LongWritable, Text, PairOfWords, IntWritable> {
	private int neighborWindow = 2;
	private PairOfWords pair = new PairOfWords();

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		String[] tokens = value.toString().split(" ");
		if (tokens.length < 2)// 不满足 两个step有单词的条件 退出
			return;
		for (String string : tokens) {
			System.out.println("the splits word of a line each time   :  " + string);
		}
		// 生成业务所需的键值对
		for (int i = 0; i < tokens.length; i++) {
			// 循环外层
			String word = tokens[i];
			pair.setWord(word);// 设置组合键的自然键 即第一个单词
			int start = (i - neighborWindow < 0) ? 0 : i - neighborWindow;
			int end = (i + neighborWindow >= tokens.length) ? tokens.length - 1 : i + neighborWindow;
			// 循环里层
			for (int j = start; j <= end; j++) {
				if (j == i)
					continue;// 跳过自身单词
				pair.setNeighbor(tokens[j]);
				System.out.println("里层  maper context 即将输出的键值对 key=" + pair);
				context.write(pair, new IntWritable(1));
			}
			// 记录当前这个单词的结伴单词数
			pair.setNeighbor("*");
			int total = end - start;
			System.out.println("外层 maper context 即将输出的键值对 key=" + pair);
			context.write(pair, new IntWritable(total));
		}
	}

	@Override
	protected void setup(Mapper<LongWritable, Text, PairOfWords, IntWritable>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		this.neighborWindow = context.getConfiguration().getInt("neighbor.window", 2);// 若没有取到该值则默认为2
	}

}
